  /*
  TP3 Algorithmes Parallèles de Tri:
  Tri à Bulles:
  p processeurs utilisés pour une liste àtrier de taille n.
  */

#include "arch.h"
#include <stdlib.h>

//user program
Global_id gid1;
Global_id gid2;
Coll_id cid1,cid2;


//la taille de la talbe
#define N 8
//nb de processeur
#define P 4
//valeur max du nombre aléatoire tiré
#define MYRAND_MAX 500

void init_gid(int, char **)
	{
		gid1.set();
		gid2.set();
		cid1.set();
		cid2.set();
		fprintf(Sdf::trace, "\nnode%d: gid1.tag=%d, gid2.tag=%d, cid1.tag=%d, cid2.tag=%d\n",
						Sdf::PROC_ID, gid1.get(), gid2.get(), cid1.get(),cid2.get());
		fflush(Sdf::trace);
	}

//definir les canaux pour communication
ChanLD<int> *chan[4];

class Tableau
	{
		public:
			int table[N/P];
	};

ChanRD<Tableau> *proc_to_left;
ChanRD<Tableau> *proc_to_right;



int compare (const void * a, const void * b)
	{
		return ( *(int*)a - *(int*)b );
	}
int flag_send;
int flag_recv;

int proc_id;
Tableau table_dist,table_locale;
int left_id, right_id;


class Calcul: public S_Thread
{
	public:
	int table_fusion[2*N/P];
	Calcul():S_Thread((Smf0_ptr)&Calcul::body)
		{
			proc_id= Sdf::PROC_ID;
		}

	void body()
		{
							for(int k = 0; k < P; k++){
/*processeur pair	*/
								if(!(proc_id%2)){

/*PHASE 1*/				if(proc_id < P -1){

                  // Emmission sur un canal local (pour synchro)
                   chan[Sdf::SND]->send(right_id);
                   chan[Sdf::RCV]->send(left_id);

										//attendre la fin du recv
										while(!flag_recv)
										{ 
										  reschedule();
										}

									//interclassertable et t dans m;
									for(int j=0; j<N/P;j++)
									{
										table_fusion[j]=table_locale.table[j];
										table_fusion[j+N/P]=table_dist.table[j];
									}

							//Attendre la fin du send
							if(!(flag_send))
							{
										while(flag_send!=1)
										{ 
										  reschedule();
										}
							}

									//table = moitié basse de m;
									qsort (table_fusion, 2*N/P , sizeof(int), compare);
									for(int j=0; j<N/P;j++)
									{
										table_locale.table[j]=table_fusion[j];
									}
								}

/***********************/
										fprintf(Sdf::trace, "node %d , Fin itération %d  de la Phase 1 : \t", proc_id,k);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/

Sdf::barrier(cid1);	

/*PHASE 2*/				if(proc_id > 0){

                  // Emmission sur un canal local (pour synchro)
                   chan[Sdf::SND]->send(left_id);
                   chan[Sdf::RCV]->send(right_id);


										//attendre la fin du recv
										while(!flag_recv)
										{ 
										  reschedule();
										}

									//inter classer table et t dans m;
									for(int j=0; j<N/P;j++)
									{
										table_fusion[j]=table_locale.table[j];
										table_fusion[j+N/P]=table_dist.table[j];
									}

							//Attendre la fin du send
							if(!(flag_send))
							{
										while(flag_send!=1)
										{ 
										  reschedule();
										}
							}

									//table = moitié haute de m;
									qsort (table_fusion, 2*N/P , sizeof(int), compare);
									for(int j=0; j<N/P;j++)
									{
										table_locale.table[j]=table_fusion[j+N/P];
									}
								}

/***********************/
										fprintf(Sdf::trace, "node %d , Fin itération %d  de la Phase 2 : \t", proc_id,k);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/

								//attendrela fin de réception;
								 Sdf::barrier(cid2);	
							}
/*processeur impair	*/
								else
								{
/*PHASE 1*/
                  // Emmission sur un canal local (pour synchro)
                   chan[Sdf::SND]->send(left_id);
                   chan[Sdf::RCV]->send(right_id);

										//attendre la fin du recv
										while(!flag_recv)
										{ 
										  reschedule();
										}

									//inter classer table et t dans m;
									for(int j=0; j<N/P;j++)
									{
										table_fusion[j]=table_locale.table[j];
										table_fusion[j+N/P]=table_dist.table[j];
									}

							//Attendre la fin du send
							if(!(flag_send))
							{
										while(flag_send!=1)
										{ 
										  reschedule();
										}
							}
									//table = moitié haute de m;
									qsort (table_fusion, 2*N/P , sizeof(int), compare);
									for(int j=0; j<N/P;j++)
									{
										table_locale.table[j]=table_fusion[j+N/P];
									}

/***********************/
										fprintf(Sdf::trace, "node %d , Fin itération %d  de la Phase 1 : \t", proc_id,k);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/

									//attendrela fin de réception;
									 Sdf::barrier(cid1);

/*PHASE 2*/				if(proc_id < P -1){

                  // Emmission sur un canal local (pour synchro)
                   chan[Sdf::SND]->send(right_id);
                   chan[Sdf::RCV]->send(left_id);

										//attendre la fin du recv
										while(!flag_recv)
										{ 
										  reschedule();
										}

									//interclassertable et t dans m;
									for(int j=0; j<N/P;j++)
									{
										table_fusion[j]=table_locale.table[j];
										table_fusion[j+N/P]=table_dist.table[j];
									}

							//Attendre la fin du send
							if(!(flag_send))
							{
										while(flag_send!=1)
										{ 
										  reschedule();
										}
							}
									//table = moitié basse de m;
									qsort (table_fusion, 2*N/P , sizeof(int), compare);
									for(int j=0; j<N/P;j++)
									{
										table_locale.table[j]=table_fusion[j];
									}
								}
/***********************/
										fprintf(Sdf::trace, "node %d , Fin itération %d  de la Phase 2 : \t", proc_id,k);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/
					//attendrela fin de réception;
					Sdf::barrier(cid2);
							}
					}
					

/***********************/
										fprintf(Sdf::trace, "node %d , TERMINE : \t", proc_id);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/

                  // Emmission sur un canal local (pour arret)
                   chan[Sdf::SND]->send(-2);
                   chan[Sdf::RCV]->send(-2);
reschedule();
}

};


class Emmission: public S_Thread{
   public:
        Emmission(): S_Thread((Smf0_ptr)&Emmission::body){       }

        void body()
					{	
						
						int temp;

for(int k = 0; k < 2*P; k++)
{
						temp = 0 ;
						flag_send = 0;
						
						chan[Sdf::SND]->recv(temp);

							if (temp == left_id)
							{
									//envoyer table àproc. proc_id-1;
									proc_to_left->send(table_locale);
									flag_send=1;

							}
							if (temp == right_id)
							{
									//envoyer table àproc. proc_id+1;
									proc_to_right->send(table_locale);
								 flag_send=1;	
							}
						
if(temp!=-2)
								reschedule();
}

}
};

class Reception: public S_Thread{
   public:

        Reception(): S_Thread((Smf0_ptr)&Reception::body){		      }

        void body()
{
int temp;
		for(int k = 0; k < 2*P; k++)
		{
							temp = 0 ;
							flag_recv=0;
							chan[Sdf::RCV]->recv(temp);
		
			if (temp == left_id)
			{
					//recevoirdans t la donnée table envoyée par proc. proc_id+1;
					proc_to_left->recv(table_dist);
					flag_recv=1;
			}
			if (temp == right_id)
			{		//recevoirdans t la donnée table envoyée par proc. proc_id-1;
					proc_to_right->recv(table_dist);
					flag_recv=1;
			}
if(temp!=-2)
		reschedule();
		}
}

};

class Root_proc: public Thread{

   public:
        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body)
								{
										proc_id= Sdf::PROC_ID;
										flag_send = 0;
								}

        void body(){

				/* Génération des nombres pseudo-aléatoirement*/
								srand (time (NULL));

								for(int j=0; j<N/P; j++)
								{
										table_locale.table[j] = rand() %MYRAND_MAX;
								}

/***********************/
										fprintf(Sdf::trace, "node %d , Remplissage de la table locale : \t", proc_id);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/

										qsort (table_locale.table, N/P , sizeof(int), compare);

/***********************/
										fprintf(Sdf::trace, "node %d , Triage de la table locale : \t", proc_id);
										fflush(Sdf::trace);

								for(int j=0; j<N/P; j++)
									{
										fprintf(Sdf::trace, "%d\t", table_locale.table[j]);
										fflush(Sdf::trace);
									}
										fprintf(Sdf::trace, "\n");
					 					fflush(Sdf::trace);
/*************************/
                left_id = ((proc_id > 0)? proc_id-1 : Sdf::NOPID);
								right_id = ((proc_id < Sdf::PROC_NB-1)? proc_id+1 : Sdf::NOPID);

                proc_to_left = new ChanRD<Tableau>(gid1, left_id);
                proc_to_right = new ChanRD<Tableau>(gid2, right_id);
								
								Sdf::barrier(cid1);

								chan[Sdf::RCV] = new ChanLD<int>;
								chan[Sdf::SND] = new ChanLD<int>;

                Calcul *calcul 		= new Calcul();
                Emmission *emmission 	= new Emmission();
                Reception *reception = new Reception();

                S_Thread *proc_tabl[3] = {calcul, emmission, reception};
                int pri[3] = {1, 1, 1};
                par(3, proc_tabl, pri);

	#if TRACE_FILE == 2
		//These functions are library internal ones,
		//they are only allowed in library test mode
		//in this case, the trace is generated in the file ARCH_trace.
		Sched_intern::sched_data->sched_q_print(1);
		Sched_intern::sched_data->embryo_q_print();
		Sched_intern::sched_data->threadctrl_q_print();
	#endif
        }
};

//end of user program
MAIN(Root_proc, init_gid, Sdf::empty, Sdf::empty)
